#include "config.h"
#include "offset_table.h"
#include "pool.h"
#include "serializer.h"
#include "log.h"
#include "query_helper.h"
#include <assert.h>
#include "monitor.h"
using namespace std;
using namespace mysqlpp;

//#define PRINTSIZE LOG << "size = " << this->size << ENDL;
#define PRINTSIZE
//row block describes the size of a contiguous set of blocks.
//numRows: number of rows belonging to this rowBlock.
//size: the actual total size of the block in bytes.
//RowBlock represents the max amount of data that can be fetched from the DB and then converted to string.

typedef long long int L;
typedef unsigned long long int UL;
#undef MAX
#define MAX(a,b)        ((a) > (b) ? (a) : (b))
#undef MIN
#define MIN(a,b)        ((a) < (b) ? (a) : (b))
struct RowBlock {	
	RowBlock* next, *prev;
	RowBlock(L so, L fr, L fr_size) {
		firstRow = lastRow = fr;
		size = fr_size;
		numRows = 1;
		next = prev = NULL;
		startOffset = so;
	};		
	bool appendRow(UL rowid, UL rowSize);
	UL firstRow, lastRow, startOffset;
	UL size, numRows;
	void print(FILE* out, int indent = 0) {
		print_indent(out, indent);
		fprintf(out, "<RowBlock> firstRow=%llu, lastRow=%llu, numRows=%llu, size=%llu, startOffset=%llu\n",
				firstRow,lastRow,numRows,size, startOffset);
	}
	void updateRowSize(UL rowid, L diff) {
		assert(rowid >=firstRow && rowid <= lastRow);
		size += diff;
	}
	void truncate(UL rowid, UL newSize) {
		size = newSize;
	}
};
//add a new row of size rowSize to this rowBlock.
//return true if the size of the rowBlock doesn't exceed the allowed limit when this row is added.
//return false otherwise.
bool RowBlock::appendRow(UL rowid, UL rowSize) {
	//LOG << "rowSize = " << rowSize << ", size = " << this->size << endl;
	assert(rowid > lastRow);
	if(this->size + rowSize > ROW_BLOCK_SIZE) {
		//LOG << "this rowBlock is full" << endl;
		return false;
	}
	++numRows;
	lastRow = rowid;
	this->size += rowSize;	
	//LOG << "numRows = " << numRows << ", size = " << numRows << endl;
	return true;
}
//OffsetBlock represents a collection of row blocks.
class OffsetBlock {
public:
	OffsetBlock *next, *prev;	
	OffsetBlock(UL so, UL fr, UL fr_size) : offsetMap(), rowBlockList() {
		firstRow = lastRow = fr;
		first = last = new RowBlock(0, fr, fr_size);
		next = prev = NULL;
		size = fr_size;
		numRows = 1;
		offsetMap[0] = last;
		startOffset = so;
		rowBlockList.push_back(last);
	};	
	//finds the rowblock that this offset lies within
	RowOffset findRows(UL offset, UL size);
	//used by the constructor to add a row.
	//if row blocks are stored as meta data it can be used later.
	//return false if there is no space to add another block
	bool appendRow(UL rowid, UL rowSize);
	~OffsetBlock();
	RowBlock *first, *last;
	//from offset within the OffsetBlock to offsetMap.
	map<UL,RowBlock*> offsetMap;
	UL startOffset, size;
	UL numRows;
	UL firstRow, lastRow;
	vector<RowBlock*> rowBlockList;
	void print(FILE *out, int indent = 0) {
		print_indent(out, indent);
		fprintf(out, "<OffsetBlock>firstRow=%llu, lastRow=%llu, numRows=%llu, size=%llu, startOffset=%llu\n",
				firstRow, lastRow, numRows, size, startOffset);
		RowBlock *tmp = first;
		while(tmp != NULL) {
			tmp->print(out, indent + 3);
			tmp = tmp->next;
		}
		print_indent(out, indent);
		fprintf(out,"</OffsetBlock>\n");
	}
	void updateRowSize(UL rowid, L diff) {
		size_t min(0), max(rowBlockList.size()), mid(0);
		while(true) {
			if(min == max) {
				mid = min;
				break;
			}
			mid = min + (max-min)/2;
			if(rowBlockList[mid]->lastRow < rowid) {
				min = mid + 1;
			} else if(rowBlockList[mid]->firstRow > rowid) {
				max = mid - 1;
			} else {
				break;
			}
		}

		rowBlockList[mid]->updateRowSize(rowid, diff);
		RowBlock *tmp = rowBlockList[mid]->next;
		while(tmp != NULL) {
			offsetMap.erase(tmp->startOffset);
			tmp->startOffset += diff;
			offsetMap[tmp->startOffset] = tmp;
			tmp = tmp->next;
		}
		this->size += diff;
	}
	void truncate(UL rowid, UL newSize);
};
bool OffsetBlock::appendRow(UL rowid, UL rowSize) {
	//LOG << "rowSize = " << rowSize << ", size = " << size << endl;
	if(size + rowSize > OFFSET_BLOCK_SIZE) {
		//LOG << "this offsetBlock is full" << endl;
		return false;
	}
	if(!last->appendRow(rowid, rowSize)) {
		RowBlock* tmp = last;
		last = new RowBlock(size, rowid, rowSize);
		tmp->next = last;
		last->prev = tmp;
		offsetMap[size] = last;
		rowBlockList.push_back(last);
	}
	lastRow = rowid;
	++numRows;
	size += rowSize;
	return true;
}

RowOffset OffsetBlock::findRows(UL offset, UL size) {
	//LOG << "ofset = " << offset << ", size = " << size << endl;
	//last row block whose offset is smaller than the given offset	
	map<UL, RowBlock*>::iterator it = offsetMap.lower_bound(offset);
#ifdef DEBUG
	if(it->first > offset && it == offsetMap.begin()) {
		LOG_ERR << "END OF THE WORLD!!" << ENDL;
		exit(1);
	}
#endif
	if(it == offsetMap.end() || (it->first > offset))
		--it;
	//LOG << "found lower offset = " <<  it->first << ENDL;
	//the rowOffset from the beginning of the rows will be offset - (offset of rowBlock)
	UL rowOffset = offset - it->first;
	//LOG << "rowOffset = " << rowOffset << ENDL;
	RowBlock* rb = it->second;
	//LOG << "rb->firstRow = " << rb->firstRow << ", rb->numRows=" << rb->numRows << ", rb->size=" << rb->size << ENDL;
#ifdef DEBUG
	if(rb->size - rowOffset < 0) {
		LOG_ERR << "SHOULD NEVER HAPPEN" << ENDL;
		exit(1);
	}
#endif
	UL curSize = MIN((UL)(rb->size - rowOffset), size);
	UL numRows = rb->numRows;
	UL firstRow = rb->firstRow;
	UL lastRow = rb->lastRow;
	rb = rb->next;
	while(rb != NULL && curSize < size) {
		//LOG << "rb->firstRow = " << rb->firstRow << ", rb->numRows="
			//<< rb->numRows << ", rb->size= " << rb->size << ENDL;
		curSize = MIN(curSize + rb->size, size);
		numRows += rb->numRows;
		lastRow = rb->lastRow;
		rb = rb->next;
	}
	//LOG << "rowOffset = " << rowOffset << ", firstRow = " << firstRow << ", numRows = "
		//<< numRows << ", curSize = " << curSize << ENDL;
	return RowOffset(rowOffset, firstRow, lastRow, numRows, curSize);
}

OffsetBlock::~OffsetBlock() {
	//LOG << ENDL;
	RowBlock* ptr = first;
	if(first == last) {
		delete first;
		return;
	}
	while(ptr != NULL) {
		RowBlock* tmp = ptr;
		ptr = ptr->next;
		delete tmp;
	}
}

map<string, OffsetTable*> OffsetTable::offsetTables;

LOCK OffsetTable::offsetTablesLock;
OffsetTable* OffsetTable::Get(const string& table, bool read_table) {
	SRLock _srl(offsetTablesLock);
	OffsetTable* ot = NULL;
	if(offsetTables.find(table) == offsetTables.end()) {
		ot = new OffsetTable(table, read_table);
	} else {
		ot = offsetTables[table];
	}
	return ot;		
}	


OffsetTable::OffsetTable(const string& table, bool read_table) :  offsetBlockList(), offsetMap(), lock() {
	WLOCK;
	{
		SRLock _srl(offsetTablesLock);
		offsetTables[table] = this;
	}
	sprintf(index_file, "/tmp/sqlfslog/%s.index", table.c_str());
	if(::truncate(index_file, 0)) {
		fprintf(stderr, "problem truncating index file %s\n", index_file);
	}
	first = last = NULL;
	numRows = 0;
	size = 0;
	firstRow = lastRow = 0;
	addAutoIncColumn(table);
	if(read_table) {
		init(table);
	}
	//print();
}

void OffsetTable::addAutoIncColumn(const string& table) {
	WLOCK;
	PRINTSIZE;
	DB_INIT;
	Connection *con = Pool::Get()->grab();
	DB_START;
	Query q = con->query();
	q << "DESCRIBE " << table;
	UseQueryResult res = q.use();
	Row row = res.fetch_row();
	string autoinc = row[5].c_str();
	while(res.fetch_row()) {}
	DB_STOP;
	if(autoinc != "auto_increment") {
		q = con->query();
		if(row[0] == "rowid") {
			DB_START;
			q << "ALTER TABLE " << table << " DROP COLUMN rowid";
			q.exec();
			DB_STOP;
			q.reset();
			(*_log) << "query = " << q.str() << ENDL;
		}
		DB_START;
		q << "ALTER TABLE " << table << " ADD COLUMN rowid INT AUTO_INCREMENT PRIMARY KEY FIRST";
		(*_log) << "query = " << q.str() << ENDL;
		q.execute();
		DB_STOP;
	}
	Pool::Get()->release(con);
}


//read the entire table from the database, or read the metadata.
void OffsetTable::init(const string& table) {
	WLOCK;
	PRINTSIZE;
//	DB_INIT;
//	TREE_INIT;
	Connection* con = Pool::Get()->grab();
	Query q = con->query();
//	DB_START;
	q << "SELECT * FROM " << table << " ORDER BY rowid";
	UseQueryResult res = q.use();
//	DB_STOP;
	while(true) {
//		DB_START;
		Row row = res.fetch_row();
//		DB_STOP;
		if(!row)
			break;
		L rowSize = Serializer::RowSize(row, 1);
//		TREE_START;
		appendRow(row[0], rowSize);
//		TREE_STOP;
	}
	Pool::Get()->release(con);
}

void OffsetTable::updateRowSize(size_t rowid, size_t old_size, size_t new_size) {
	WLOCK;
	PRINTSIZE;
	if(rowid > lastRow) {
		assert(old_size == 0);
		this->appendRow(rowid, new_size);
		return;
	}
	//binary search to find the OffsetBlock that contains rowid
	size_t min(0), max(offsetBlockList.size()), mid(0);
	while(true) {
		if(min == max) {
			mid = min;
			break;
		}
		mid = min + (max-min)/2;
		if(offsetBlockList[mid]->lastRow < rowid) {
			min = mid + 1;
		} else if(offsetBlockList[mid]->firstRow > rowid) {
			max = mid - 1;
		} else {
			break;
		}
	}
	L diff = (L)new_size - (L)old_size;
	offsetBlockList[mid]->updateRowSize(rowid, diff);
	OffsetBlock *tmp = offsetBlockList[mid]->next;
	while(tmp != NULL) {
		offsetMap.erase(tmp->startOffset);
		tmp->startOffset += diff;
		offsetMap[tmp->startOffset] = tmp;
		tmp = tmp->next;
	}
	this->size += diff;
}
void OffsetTable::appendRow(UL rowid, UL rowSize) {
	WLOCK;
	PRINTSIZE;
	if(last == NULL) {
		first = last = new OffsetBlock(0, rowid, rowSize);
		offsetMap[0] = first;
		size = rowSize;
		numRows = 1;
		firstRow = lastRow = rowid;
		offsetBlockList.push_back(last);
		return;
	}
	//printf("rowid = %llu, rowSize=%llu\n", rowid, rowSize);
	assert(rowid >lastRow);
	if(!last->appendRow(rowid, rowSize)) {
		//LOG << "offsetBlock full, starting a new one at offset " << size << ENDL;
		OffsetBlock* tmp = last;
		last = new OffsetBlock(size, rowid, rowSize);
		tmp->next = last;
		last->prev = tmp;
		offsetMap[size] = last;
		offsetBlockList.push_back(last);
	}
	//LOG << "rowSize = " <<  rowSize << ENDL;
	size += (unsigned long)rowSize;
	++numRows;
	lastRow = rowid;
}



RowOffset OffsetTable::findRows(UL offset, UL size) {
	RLOCK;
	PRINTSIZE;
	//LOG << "offset = " << offset << " size=" << size << ENDL;
	if(this->size == 0) {
		assert(offset == 0);
		return RowOffset(0, 0, 0, 0, 0);
	}
	map<UL, OffsetBlock*>::iterator it = offsetMap.lower_bound(offset);
#ifdef DEBUG
	if(it->first > offset && it == offsetMap.begin()) {
		LOG_ERR << "END OF THE WORLD!!" << ENDL;
		exit(1);
	}
#endif
	if(it == offsetMap.end() || (it->first > offset))
		--it;
	L reducedOffset = offset - it->first;
	//LOG << "the beginning of the offsetBlock = " <<  it->first << ENDL;
	//LOG << "reducedOFfset = " << reducedOffset << ENDL;
	OffsetBlock* ob = it->second;
	RowOffset ro = ob->findRows(reducedOffset, size);
	UL resOffset = ro.offset;
	UL firstRow = ro.firstRow;
	UL lastRow  = ro.lastRow;
	UL numRows = ro.numRows;
	UL remSize = size - ro.size;
	ob = ob->next;
	while(ob!= NULL && remSize > 0) {
		ro = ob->findRows(0, remSize);
		remSize -= ro.size;
		numRows += ro.numRows;
		lastRow = ro.lastRow;
		ob = ob->next;
	}
	return RowOffset(resOffset, firstRow, lastRow, numRows, size-remSize);
}

void OffsetBlock::truncate(UL rowid, UL newSize) {
	RowBlock* ptr = first;
	size = 0;
	numRows = 0;
	last = first;
	rowBlockList.clear();
	while(ptr != NULL) {
		if(rowid > ptr->lastRow) {
			size += ptr->size;
			newSize -= ptr->size;
			numRows += ptr->numRows;
			last = ptr;
			rowBlockList.push_back(ptr);
			ptr = ptr->next;
			continue;
		}
		if(rowid >= ptr->firstRow && rowid <= ptr->lastRow) {
			ptr->truncate(rowid, newSize);
			size += ptr->size;
			numRows += ptr->numRows;
			last = ptr;
			rowBlockList.push_back(ptr);
			ptr = ptr->next;
			continue;
		}
		if(rowid < ptr->firstRow) {
			RowBlock *tmp = ptr;
			ptr = ptr->next;
			delete tmp;
		}
	}
	last->next = NULL;
}

void OffsetTable::truncate(UL rowid, UL newSize) {
	WLOCK;
	PRINTSIZE;
	OffsetBlock *ptr = first;
	size = 0;
	numRows = 0;
	last = first;
	offsetBlockList.clear();
	while(ptr != NULL) {
		if(rowid > ptr->lastRow) {
			size += ptr->size;
			newSize -= ptr->size;
			numRows += ptr->numRows;
			last = ptr;
			offsetBlockList.push_back(ptr);
			ptr = ptr->next;
			continue;
		}
		if(rowid >= ptr->firstRow && rowid <= ptr->lastRow) {
			ptr->truncate(rowid, newSize);
			size += ptr->size;
			numRows += ptr->numRows;
			last = ptr;
			offsetBlockList.push_back(ptr);
			ptr = ptr->next;
			continue;
		}
		if(rowid < ptr->firstRow) {
			OffsetBlock *tmp = ptr;
			ptr = ptr->next;
			delete tmp;
		}

	}
	last->next = NULL;
}
void OffsetTable::clear() {
	WLOCK;
	PRINTSIZE;
	//LOG << ENDL;
	OffsetBlock *ptr = first;
	offsetBlockList.clear();
	if(first == last) {
		delete first;
		ptr = NULL;
	}
	while(ptr != NULL) {
		OffsetBlock *tmp = ptr;
		ptr = ptr->next;
		delete tmp;
	}
	offsetMap.clear();
	first = last = NULL;
	numRows = 0;
	size = 0;
	firstRow = lastRow = 0;
}
OffsetTable::~OffsetTable() {
	WLOCK;
	PRINTSIZE;
	clear();
}

void OffsetTable::print(FILE *out, int indent) {
	RLOCK;
	bool close = out == NULL;
	if(out == NULL)
		out = fopen(index_file, "a");
	print_indent(out, indent);
	fprintf(out, "<OffsetTable> firstRow=%llu, lastRow=%llu, numRows=%llu, size=%llu\n",
					firstRow, lastRow, numRows, size);
	OffsetBlock *tmp = first;
	while(tmp!= NULL) {
		tmp->print(out, indent + 3);
		tmp = tmp->next;
	}
	print_indent(out, indent);
	fprintf(out, "</OffsetTable>\n");
	fprintf(out,"\n================================\n");
	if(close)
		fclose(out);
}

UL OffsetTable::getSize() {
	RLOCK;
	return this->size;
}

UL OffsetTable::getLastRow() {
	RLOCK;
	return this->lastRow;
}


void print_indent(FILE* out, int indent) {
	for(int i = 0; i < indent; ++i) {
		if((i %3) == 0)
			fprintf(out, "|");
		else
			fprintf(out, " ");
	}
}
